package org.example.com.atguigu.exercise;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Int;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class Exercise1 {
    public static void main(String[] args) {
        // 统计菜的种类数最多的三个省份
        /*
        0       1       2           3                           4   5
        香菇	    12.00	2018/1/1	四川南充市桑园坝农产品批发市场	四川	南充
        油菜		        2018/1/1
        大白菜	0.40	2018/1/1	安徽砀山农产品中心惠丰批发市场	安徽	砀山
        * */
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // rdd1 [(西红柿	5.20	2018/1/1	四川南充市桑园坝农产品批发市场	四川	南充), ...]
        JavaRDD<String> rdd0 = sc.textFile("E:\\大数据第三阶段\\spark\\6.代码\\product.txt");
        // rdd2 [ 蛇果	山东, 柠檬	山东, 椰子	山东, 甜瓜	山东]
        JavaRDD<String> rdd1 = rdd0.map(new Function<String, String>() {
            @Override
            public String call(String v1) throws Exception {
                if (v1.split("\t").length == 6) {
                    String s = "" + v1.split("\t")[0] + "\t" + v1.split("\t")[4];
                    return s;
                } else return null;
            }
        });
        // 去重!!!
        JavaRDD<String> rdd2 = rdd1.distinct();

        // rdd3 [(辽宁, 韭菜 豆角...), (安徽, 西红柿 尖椒...)(...)...]
        JavaPairRDD<String, Iterable<String>> rdd3 = rdd2.groupBy(new Function<String, String>() {
            @Override
            public String call(String v1) throws Exception {
                if (v1!=null){
                    String[] split = v1.split("\t");
                    return split[1];
                }else return null;
            }
        });
        // rdd4 去脏数据
        JavaPairRDD<String, Iterable<String>> rdd4 = rdd3.filter(new Function<Tuple2<String, Iterable<String>>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Iterable<String>> v1) throws Exception {
                return v1._1 != null;
            }
        });
        // rdd5 [(457,北京), (361,山西), (354,江苏)...]
        JavaPairRDD<Integer, String> rdd5 = rdd4.mapToPair(new PairFunction<Tuple2<String, Iterable<String>>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Iterable<String>> stringIterableTuple2) throws Exception {
                int length = stringIterableTuple2._2.toString().split(",").length;
                return new Tuple2<>(length, stringIterableTuple2._1);
            }
        }).sortByKey(false);
//        System.out.println(rdd5.collect());
        // 去重
        System.out.println("菜的种类数最多的三个省份:");
        List<Tuple2<Integer, String>> take = rdd5.take(3);
        Iterator<Tuple2<Integer, String>> iterator = take.iterator();
        while (iterator.hasNext()) System.out.print(iterator.next()._2+" ");
        rdd5.saveAsTextFile("output");
    }
}
